use std::cell::RefCell;
use std::collections::{HashMap, HashSet, VecDeque};
use chrono::Local;
use yewdux::Dispatch;
use crate::business::app_state::AppState;
use crate::business::page::fragment::im_session_list_view::{SessionAction, SessionListState};
use crate::business::rpc::service::rpc_message_service::RpcMessageService;
use crate::business::websocket::dto::im_message_dto::MessageReadReq;
use crate::db::dao::im_message_dao::{change_message_action_status2, change_message_action_status3};
use crate::db::dao::im_session_dao::incr_hot_message_num;
use crate::db::db::Db;
use crate::db::pojo::pojo::{ImMessage, ImMessageActionStatus, ImMessageFileStatus, ImMessageStatus};

const CACHE_TIME: usize = 5 * 60 * 1000;  // 缓存时间
const BATCH_SIZE: usize = 200;

pub struct MessageReadService {
    queue: RefCell<VecDeque<MessageReadData>>,
    read_map: RefCell<HashMap<String, u64>>,
    user_id: String,
    app_state: AppState,
    session_dispatch: Dispatch<SessionListState>,
}

impl MessageReadService {

    pub fn new(user_id: String, app_state: AppState, dispatch: Dispatch<SessionListState>) -> MessageReadService{
        MessageReadService {
            queue: RefCell::new(VecDeque::new()),
            read_map: RefCell::new(HashMap::new()),
            user_id,
            app_state,
            session_dispatch: dispatch,
        }
    }

    pub async fn async_handle_read(&self, user_id: String, message_read_data_list: Vec<MessageReadData>) {
        log::info!("do read message business0, data: {:?}", message_read_data_list.len());
        let need_handle_messages = self.filter_need_handle_messages(&user_id, message_read_data_list);
        for need_handle_message in need_handle_messages {
            self.queue.borrow_mut().push_back(need_handle_message);
        }

        self.do_read_business().await;
    }

    pub async fn do_read_business(&self) {
        let need_handle_data_list = self.pull_messages(BATCH_SIZE);
        if need_handle_data_list.is_empty() {
            return;
        }
        log::info!("do read message business, data: {:?}", need_handle_data_list);
        self.do_business_read_messages(need_handle_data_list).await;
    }
    
    async fn do_business_read_messages(&self, mut need_handle_messages: Vec<MessageReadData>) {
        let mut message_server_id_set = HashSet::new();
        let mut message_id_set = HashSet::new();
        
        for message_read_data in need_handle_messages.iter() {
            message_server_id_set.insert(message_read_data.message_server_id.clone());
            
            if message_read_data.action_status != ImMessageActionStatus::Read {
                message_id_set.insert(message_read_data.message_id.clone());
            }
            
            self.read_map.borrow_mut().entry(message_read_data.message_server_id.clone()).or_insert(Local::now().timestamp().cast_unsigned());
        }

        log::info!("do read message business, message_id_set: {:?}", message_server_id_set);
        if !message_server_id_set.is_empty() {
            self.change_action_status2(message_id_set, &mut need_handle_messages, ImMessageActionStatus::Read);

            let is_success = self.send_read_action_to_server(&message_server_id_set).await;

            log::info!("do read message business, send to server result: {:?}", is_success);
            if is_success {
                self.change_action_status(message_server_id_set, &mut need_handle_messages, ImMessageActionStatus::Readed);
            }
            need_handle_messages.iter_mut().for_each(|message_read_data| {
                message_read_data.action_status = ImMessageActionStatus::Readed;
            });

            let session_size_map = need_handle_messages.iter()
                .fold(HashMap::new(), |mut map, message_read_data| {
                    *map.entry(message_read_data.session_id.clone()).or_insert(0) += 1;
                    log::info!("do read message business, session_size_map0: {:?}", map);
                    map
                });

            log::info!("do read message business, session_size_map: {:?}", session_size_map);
            
            let db = Db::open();
            for (key, value) in session_size_map.iter() {
                let s : i32 = value.clone() as i32 * (-1);
                incr_hot_message_num(db.clone(), key.clone(), s);
            }
            self.session_dispatch.clone().apply(SessionAction::DecreaseHotMessages(session_size_map));
        }
        
    }

    fn filter_need_handle_messages(&self, user_id: &str, message_read_data_list: Vec<MessageReadData>)->Vec<MessageReadData> {
        let mut result = vec![];
        for message_read_data in message_read_data_list {
            if message_read_data.is_sender(user_id) || message_read_data.message_server_id == "" {
                continue;
            }
            if message_read_data.action_status != ImMessageActionStatus::None &&
                message_read_data.action_status != ImMessageActionStatus::Read {
                continue;
            }

            let map = self.read_map.borrow();
            let time = map.get(&message_read_data.message_server_id.clone());
            if let Some(time) = time {
                let diff_time = Local::now().timestamp().cast_unsigned() - time;
                if diff_time < CACHE_TIME as u64 {
                    continue;
                }
            }
            result.push(message_read_data);
        }
        result
    }
    
    fn pull_messages(&self, size: usize) -> Vec<MessageReadData> {
        let mut result = vec![];
        for _ in 0..size {
            let message_data = self.queue.borrow_mut().pop_front();
            if let Some(message_data) = message_data {
                result.push(message_data);
            }
        }
        result
    }

    fn change_action_status2(&self, message_id_set: HashSet<u64>, need_handle_messages: &mut Vec<MessageReadData>, action_status: ImMessageActionStatus) {
        need_handle_messages.iter_mut().for_each(|message_read_data| {
            message_read_data.action_status = action_status.clone();
        });

        let db = Db::open();
        change_message_action_status2(db.clone(), message_id_set, action_status);
    }

    fn change_action_status(&self, message_server_set: HashSet<String>, need_handle_messages: &mut Vec<MessageReadData>, action_status: ImMessageActionStatus) {
        need_handle_messages.iter_mut().for_each(|message_read_data| {
            message_read_data.action_status = action_status.clone();
        });

        let db = Db::open();
        change_message_action_status3(db.clone(), &self.user_id, message_server_set, action_status);
    }

    async fn send_read_action_to_server(&self, message_server_id_set: &HashSet<String>)-> bool {
        let message_ids = message_server_id_set.iter().map(|id| id.to_string()).collect::<HashSet<String>>();
        if message_ids.len() < 1 {
            return false;
        }
        let message_req = MessageReadReq {
            message_ids,
            create_date_time: Local::now(),
            local_seq: Some("".to_string()),
        };

        log::info!("sending message read request to server: {:?}", message_req);
        let result = RpcMessageService::send_read_message(self.app_state.clone(), message_req).await;
        result.is_ok()
    }
}

#[derive(Debug)]
pub struct MessageReadData {
    message_id: u64,
    message_server_id: String,
    from_user_id: String,
    session_id: u64,
    status: ImMessageStatus,
    action_status: ImMessageActionStatus,
    file_status: ImMessageFileStatus,
}

impl From<&ImMessage> for MessageReadData {
    fn from(message: &ImMessage) -> Self {
        let message_server_id = message.server_id.clone();
        let message_server_id = if message_server_id.is_some() {
            message_server_id.unwrap()
        } else {
            "".to_string()
        };
        Self {
            message_id: message.id.unwrap(),
            message_server_id,
            from_user_id: message.from_user_id.clone(),
            session_id: message.session_id.clone(),
            status: message.status.clone(),
            action_status: message.action_status.clone(),
            file_status: message.file_status.clone(),
        }
    }
}

impl MessageReadData {

    pub fn is_sender(&self, user_id: &str)->bool {
        self.from_user_id == user_id
    }
}

